package com.niit.spark.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Date:2025/4/11
 * Author：Ys
 * Description:
 */
object CreateRDD {




  def main(args: Array[String]): Unit = {
    //1.准备环境                     调用CPU所有核数进行计算
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("CreateRDD")

    //2.让sc 去加载配置  sc==> sparkContext
    val sc = new SparkContext(sparkConf)

    //在内存中创建RDD
    memoryCreateRDD(sc)

    //在外部数据源创建RDD
    fileCreateRDD(sc)

    //3.关闭环境
    sc.stop()
  }

  def memoryCreateRDD(sc: SparkContext) = {
    //内存中创建RDD  parallelize        如果只有一个参数，采用的是默认分区数。默认CPU核数
    val rdd1: RDD[Int] = sc.parallelize(1 to 10)
    //获得rdd的底层分区数
     println(rdd1.getNumPartitions)
     println(rdd1.partitions.length)
    //将rdd保存到文件当中
    //rdd1.saveAsTextFile("output")

    val rdd2: RDD[Int] = sc.parallelize(1 to 10, 3)//该RDD的分区数为3
    //获得rdd的底层分区数
    println(rdd2.getNumPartitions)
    println(rdd2.partitions.length)
    //将rdd保存到文件当中
    //rdd2.saveAsTextFile("output1")

    //将RDD进行遍历 --> 将RDD转换成数组 在进行遍历
    rdd1.collect().foreach(println)

    //makeRDD 和 parallelize 是一样的。 parallelize是makeRDD底层写法
    val rdd3: RDD[Int] = sc.makeRDD(1 to 10)
    rdd3.collect().foreach(println)


  }
  def fileCreateRDD(sc: SparkContext) = {
    //1.在项目目录上创建一个input文件夹，并在该文件夹中创建1.txt，里面写一些东西
    val rdd3: RDD[String] = sc.textFile("input/1.txt")//会自行自动分片

    rdd3.collect().foreach(println)


  }
}
